package com.peipei.hbase.utils;

import com.google.gson.Gson;
import com.peipei.hbase.constants.Constants;
import com.peipei.hbase.model.Student;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.XAttr;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;


public class HBaseOperationUtils {
    private static final Logger logger = LoggerFactory.getLogger(HBaseOperationUtils.class);

    /**
     * connection for hbase
     */
    private static Connection connection;
    /**
     * configuration for hbase
     */
    private static Configuration configuration;
    /**
     * manager for hbase
     */
    private static Admin admin;

    static {
        configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.quorum", "192.168.1.180:2181,192.168.1.181:2181,192.168.1.182:2181");
        try {
            connection = ConnectionFactory.createConnection(configuration);
            admin = connection.getAdmin();
        } catch (IOException e) {
            logger.info("", e);
        }
    }

    private HBaseOperationUtils() {

    }

    /**
     * get htable according to name
     *
     * @param name name of htable
     * @return
     */
    public static Table getTable (String name) throws IOException {
        return connection.getTable(TableName.valueOf(name));
    }

    /**
     * get htable according to name
     *
     * @param name name of htable
     * @return
     */
    public static Table createTable (String name) throws IOException {
        return connection.getTable(TableName.valueOf(name));
    }

    /**
     * create namespace for hbase
     *
     * @param name name of namespace of hbase
     * @throws IOException
     */
    public static void createNamespace (String name) throws IOException {
        admin.createNamespace(NamespaceDescriptor.create(name).build());
    }

    /**
     * create table for hbase
     *
     * @param namespace namespace for table belongs to
     * @param table table will be created
     * @param columns columns for table
     */
    public static void createTable (String namespace, String table, String ... columns) throws IOException {
        // table descriptor
        TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(TableName.valueOf(namespace + ":" + table));

        List<ColumnFamilyDescriptor> columnFamilyDescriptorList = new ArrayList<>(columns.length);
        for (String column: columns){
            ColumnFamilyDescriptorBuilder columnFamilyDescriptorBuilder = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(column));
            columnFamilyDescriptorList.add(columnFamilyDescriptorBuilder.build());
        }

        tableDescriptorBuilder.setColumnFamilies(columnFamilyDescriptorList);
        admin.createTable(tableDescriptorBuilder.build());
    }

    /**
     * list all table descriptors
     * @return
     * @throws IOException
     */
    public static List<TableDescriptor> listTables () throws IOException {
        return admin.listTableDescriptors();
    }


    /**
     * list tables according to namespace
     *
     * @param namespace namespace
     * @return
     * @throws IOException
     */
    public static List<TableDescriptor> listTables (String namespace) throws IOException {
        return admin.listTableDescriptorsByNamespace(Bytes.toBytes(namespace));
    }

    /**
     *
     * @param tableName
     * @return
     * @throws IOException
     * @throws IllegalAccessException
     * @throws InstantiationException
     * @throws NoSuchFieldException
     */
    public static List<Map<String, Object>> listAll(String tableName) throws IOException, IllegalAccessException, InstantiationException, NoSuchFieldException {
        // get table by name
        Table table = getTable(tableName);
        //  create scan
        Scan scan = new Scan();
        // create sanner for result
        ResultScanner resultScanner = table.getScanner(scan);

        List<Map<String, Object>> list = new ArrayList<>();

        for (Result result : resultScanner){
            Map<String, Object> item = new HashMap<>();
            item.put(Constants.ROW_KEYS, result.getRow());

            Cell[] cells = result.rawCells();
            for (int i = 0; cells != null &&i < cells.length; i++){
                Cell c = cells[i];
                String columnName = Bytes.toString(c.getQualifierArray(), c.getQualifierOffset(), c.getQualifierLength());
                String columnValue = Bytes.toString(c.getValueArray(), c.getValueOffset(), c.getValueLength());

                item.put(columnName, columnValue);
            }

            list.add(item);
        }
        return list;
    }

    /**
     * get item by rowkey
     *
     * @param namespace
     * @param tableName
     * @param rowkey
     * @return
     * @throws IOException
     */
    public static Map<String, Object> getByRowKey (String namespace, String tableName, String rowkey) throws IOException {
        Table table = getTable(namespace + ":" + tableName);
        // create SCAN object
        Scan scan = new Scan();
        scan.withStartRow(Bytes.toBytes(rowkey));
        // create ReulstScanner
        ResultScanner resultScanner = table.getScanner(scan);

        for (Result result : resultScanner){
            Map<String, Object> item = new HashMap<>();
            item.put(Constants.ROW_KEYS, result.getRow());

            Cell[] cells = result.rawCells();
            for (int i = 0; cells != null &&i < cells.length; i++){
                Cell c = cells[i];
                String columnName = Bytes.toString(c.getQualifierArray(), c.getQualifierOffset(), c.getQualifierLength());
                String columnValue = Bytes.toString(c.getValueArray(), c.getValueOffset(), c.getValueLength());

                item.put(columnName, columnValue);
            }
            return item;
        }
        return null;
    }

    /**
     * delete row according to rowkey
     * @param namespace
     * @param tableName
     * @param rowKey
     * @throws IOException
     */
    public static void delete (String namespace, String tableName, String rowKey) throws IOException {
        Table table = getTable(namespace + ":" + tableName);

        Get get = new Get (Bytes.toBytes(rowKey));
        Result result = table.get(get);
        logger.info("{}", new Gson().toJson(result));

        if (table.exists(get)){

            Delete delete = new Delete(Bytes.toBytes(rowKey));
            table.delete(delete);
        }


        logger.info("{}", table.exists(new Get (Bytes.toBytes(rowKey))));




//        table.close();
    }

    /**
     * put data to hbase
     *
     * @param namespace namespace of table
     * @param tableName name of name
     * @param student data will be put to
     */
    public static void putData (String namespace, String tableName, String columnFamily, Student student) throws IOException {
        String rowkey = RowKeyUtils.getRowKey();
        Table table = HBaseOperationUtils.getTable(namespace + ":" + tableName);

        // create PUT object
        Put put = new Put(Bytes.toBytes(rowkey));

        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("id"),  Bytes.toBytes(student.getId()));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"), Bytes.toBytes(student.getName()));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("grade"), Bytes.toBytes(student.getGrade()));
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("corse"), Bytes.toBytes(student.getCorse()));

        table.put(put);
        table.close();
    }
}
